package netty.mqtt.t1;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;

class MqttServerHandler extends SimpleChannelInboundHandler<MqttMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) {
        if (msg.fixedHeader() == null) {
            ctx.close();
            return;
        }

        switch (msg.fixedHeader().messageType()) {
            case CONNECT:
                handleConnect(ctx, (MqttConnectMessage) msg);
                break;

            case SUBSCRIBE:
                handleSubscribe(ctx, (MqttSubscribeMessage) msg);
                break;

            case PUBLISH:
                handlePublish(ctx, (MqttPublishMessage) msg);
                break;

            case DISCONNECT:
                ctx.close();
                break;

            default:
                System.out.println("Unsupported message type: " + msg.fixedHeader().messageType());
        }
    }

    private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
        System.out.println("Client connected: " + msg.payload().clientIdentifier());

        MqttConnAckMessage connAckMessage = MqttMessageBuilders.connAck()
                .returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
                .build();

        ctx.writeAndFlush(connAckMessage);
    }

    private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
        System.out.println("Subscription request received.");

        MqttSubAckMessage subAckMessage = MqttMessageBuilders.subAck()
                .packetId(msg.variableHeader().messageId())
                .addGrantedQos(MqttQoS.AT_LEAST_ONCE)
                .build();

        ctx.writeAndFlush(subAckMessage);
    }

    private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
        String topic = msg.variableHeader().topicName();
        String payload = msg.payload().toString(CharsetUtil.UTF_8);

        System.out.println("Message published to topic: " + topic + ", payload: " + payload);

        // Send PUBACK for QoS 1 messages
        if (msg.fixedHeader().qosLevel() == MqttQoS.AT_LEAST_ONCE) {
            MqttMessage pubAckMessage = MqttMessageBuilders.pubAck()
                    .packetId(msg.variableHeader().packetId())
                    .build();

            ctx.writeAndFlush(pubAckMessage);
        }

    }

    @Override
    public void exceptionCaught(io.netty.channel.ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
